package com.andnnl.pool;

import com.andnnl.pool.retry.RetryMsg;
import com.andnnl.pool.retry.RetryMsgQueue;
import com.andnnl.pool.user.OneUser;
import com.andnnl.pool.user.PriorityUser;
import com.andnnl.pool.user.TwoUser;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.IntSupplier;
import java.util.stream.Collectors;

/**
 * 多用户消息轮流处理，支持运行时修改线程大小、队列大小、每个用户每次处理消息数
 * Created by chenss on 2021/6/23.
 */
public class TaskManage<T> {
    private static final Logger logger = LoggerFactory.getLogger(TaskManage.class);
    private final IntParam threadSize = new IntParam(5);//线程大小
    private final IntParam queueCapacity = new IntParam(threadSize.getValue() * 2);//队列长度
    private final UserQueue<T> userQueue = new UserQueue<>();//用户队列
    private final Map<String, IUser<T>> userMap = new ConcurrentHashMap<>();//用户map
    private IMsgHandle<T> handle = null;//handle
    private IRetryMsgFilter<T> retryMsgFilter = null;//filter
    private RetryMsgQueue<T> retryMsgQueue = null;//重试队列
    private ThreadPoolExecutor pool = null;
    private MsgTypeEnum msgTypeEnum = MsgTypeEnum.OneQueue;
    private int batchSize = 1;//每次取多少个
    private int refreshParamSec = 10;//刷新参数时间间隔(秒)

    private boolean isInitBindThread = false;//是否初始化绑定线程任务
    private final AtomicLong handleDataSize = new AtomicLong();//累计处理过msg的DataSize

    /**
     * 启动
     *
     * @return
     */
    public TaskManage start() {
        if (this.handle == null) {
            throw new IllegalArgumentException("no handle!");
        }
        if (pool == null)
            pool = new ThreadPoolExecutor(threadSize.getValue(), threadSize.getValue(), 60L, TimeUnit.MILLISECONDS,
                    new ResizableCapacityLinkedBlockingQueue<>(queueCapacity.getValue())
                    , new BlockingRejectedExecutionHandler());//添加任务阻塞线程池

        Executors.newSingleThreadExecutor().submit(() -> handleUser());
        //每分钟清理一次map里的用户对象,避免对象过多
        Executors.newSingleThreadScheduledExecutor()
                .scheduleWithFixedDelay(() -> clearExpireUser(), 60, 60, TimeUnit.SECONDS);

        if (retryMsgFilter != null) {
            retryMsgQueue = new RetryMsgQueue();
            Executors.newSingleThreadExecutor().submit(() -> handleRetryMsg());
        }
        if (isInitBindThread) {
            //每10秒检查一下参数，并更新
            Executors.newSingleThreadScheduledExecutor()
                    .scheduleWithFixedDelay(() -> refreshParam(), 10, 10, TimeUnit.SECONDS);
        }
        return this;
    }

    /**
     * 清理掉超时不用的User对象(清理3分钟前的)
     */
    private void clearExpireUser() {
        Iterator<Map.Entry<String, IUser<T>>> it = userMap.entrySet().iterator();
        while (it.hasNext()) {
            Map.Entry<String, IUser<T>> entry = it.next();
            IUser<T> user = entry.getValue();
            // 增加更严格的检查，确保用户不仅队列为空，而且不在用户队列中
            if (!user.getInQueue().get() && user.getCurrTime() < System.currentTimeMillis() - 3 * 60 * 1000L && user.isAllEmpty()) {
                it.remove();
                logger.info("remove expire use:" + user.getName());
            }
        }
    }


    public static TaskManage build() {
        return new TaskManage();
    }

    /**
     * 使用泛型
     *
     * @param <T>
     * @return
     */
    public static <T> TaskManage<T> buildGeneric() {
        return new TaskManage();
    }

    public IMsgHandle<T> getHandle() {
        return handle;
    }

    public TaskManage<T> setHandle(IMsgHandle<T> handle) {
        handle.setTaskManage(this);
        this.handle = handle;
        return this;
    }

    /**
     * 轮询用户处理
     */
    private void handleUser() {
        try {
            while (!Thread.currentThread().isInterrupted()) {
                IUser<T> user = userQueue.taskUser();
                if (user == null) continue;
                user.getInQueue().set(false);//先标记为,不在队列中
                List<Msg<T>> msgList = user.takeMsg();
                if (msgList.isEmpty()) continue;//用户没数据,直接忽略
                pool.submit(() -> {
                    try {
                        msgList.forEach(m -> handleDataSize.addAndGet(m.getDataSize()));//累计处理过msg的DataSize
                        handle.handleMsg(this, msgList);
                    } catch (Exception e) {
                        logger.error("catch Exception:" + e.getMessage(), e);
                    }
                });
                userQueue.addUser(user);//如果用户的消息队列不为空，放到队列后再处理
            }
        } catch (Exception e) {
            logger.error("handleUser thread error", e);
            Thread.currentThread().interrupt(); // 恢复中断状态
        }
    }

    /**
     * 添加延迟重试任务(也可首次添加，作为延迟消息)
     *
     * @param msg
     * @return true:添加成功
     */
    public boolean addRetryMsg(Msg<T> msg) {
                if (retryMsgFilter == null) {
                    throw new IllegalArgumentException("no retryMsgFilter! can not addRetryTask");
                }
                Objects.requireNonNull(msg, "msg is null");
                Objects.requireNonNull(msg.getUserName(), "msg.getUserName() is null");

                RetryMsg<T> retryMsg = new RetryMsg<>(msg);
                if (retryMsgFilter.filterDelayMsg(retryMsg)) {
                    retryMsgQueue.addRetryTask(retryMsg);
                    return true;
                }
                return false;
            }

    /**
     * 处理重试任务
     */
    private void handleRetryMsg() {
        try {
            while (!Thread.currentThread().isInterrupted()) {
            RetryMsg<T> retryMsg = retryMsgQueue.taskRetryMsg();
            if (retryMsg == null) continue;
            //到达任务延迟时间后，自动添加到任务列表
            IUser<T> user = retryMsg.getMsg().getUser();
            String name = user != null ? user.getName() : retryMsg.getMsg().getUserName();
            addMsg(name, retryMsg.getMsg());
            }
        } catch (Exception e) {
            logger.error("handleRetryMsg thread error", e);
            Thread.currentThread().interrupt(); // 恢复中断状态
        }
    }

    public int getThreadSize() {
        return threadSize.getValue();
    }

    /**
     * 运行时，改变线程池大小，注意：需要核心线程数与最大线程数一致才会生效
     *
     * @param threadSize
     * @return
     */
    public TaskManage setThreadSize(int threadSize) {
        this.threadSize.setValue(threadSize);
        if (pool != null) {
            pool.setCorePoolSize(threadSize);
            pool.setMaximumPoolSize(threadSize);
            pool.prestartAllCoreThreads();
        }
        return this;
    }

    public int getQueueCapacity() {
        return queueCapacity.getValue();
    }

    /**
     * 动态修改队列大小(不建议随意更改，有可能影响正在执行的任务，导致任务卡死)
     *
     * @param queueCapacity
     * @return
     */
    @Deprecated
    public TaskManage setQueueCapacity(int queueCapacity) {
        this.queueCapacity.setValue(queueCapacity);
        if (pool != null) {
            BlockingQueue<Runnable> q = pool.getQueue();
            if (q instanceof ResizableCapacityLinkedBlockingQueue) {
                ((ResizableCapacityLinkedBlockingQueue<Runnable>) q).setCapacity(queueCapacity);
            }
        }
        return this;
    }


    public ThreadPoolExecutor getPool() {
        return pool;
    }

    /**
     * 添加自定义线程池
     *
     * @param pool
     * @return
     */
    public TaskManage setPool(ThreadPoolExecutor pool) {
        this.pool = pool;
        return this;
    }


    /**
     * 返回当前线程池的状态信息
     *
     * @param pool
     * @param name
     * @return
     */
    public static String threadPoolStatus(ThreadPoolExecutor pool, String name) {
        BlockingQueue queue = pool.getQueue();
        String x = name + "-:" +
                " 核心线程数:" + pool.getCorePoolSize() +
                " 活动线程数:" + pool.getActiveCount() +
                " 最大线程数:" + pool.getMaximumPoolSize() +
                " 线程池活跃度:" +
                divide(pool.getActiveCount(), pool.getMaximumPoolSize()) +
                " 任务完成数:" + pool.getCompletedTaskCount() +
                " 队列大小:" + (queue.size() + queue.remainingCapacity()) +
                " 当前排队线程数:" + queue.size() +
                " 队列剩余大小:" + queue.remainingCapacity() +
                " 队列使用度:" + divide(queue.size(), queue.size() + queue.remainingCapacity());
        return x;
    }

    private static String divide(int num1, int num2) {
        return String.format("%1.2f%%",
                Double.parseDouble(num1 + "") / Double.parseDouble(num2 + "") * 100);
    }

    /**
     * 返回所有的用户数和等待消息数
     *
     * @return
     */
    public String getMsgSizeStr() {
        Set<Map.Entry<String, IUser<T>>> es = getUserMap().entrySet();
        String s = " 用户数:" + es.size() +
                " 总消息:" + es.stream().collect(Collectors.summingLong(m -> m.getValue().getTotalSize())) +
                " 剩余DataSize:" + es.stream().collect(Collectors.summingLong(m -> m.getValue().getTotalDataSize())) +
                " 处理过的DataSize:" + handleDataSize.get() +
                " userQueue:" + userQueue.getQueueSize();
        if (retryMsgQueue != null) {
            s += " 重试消息:" + retryMsgQueue.getSize();
        }
        return s;
    }

    /**
     * 返回总消息数（不包括重试数）
     *
     * @return
     */
    public long getMsgSize() {
        Set<Map.Entry<String, IUser<T>>> es = getUserMap().entrySet();
        return es.stream().collect(Collectors.summingLong(m -> m.getValue().getTotalSize()));
    }

    /**
     * 返回总消息数,每个Msg的dataSize（不包括重试数）
     *
     * @return
     */
    public long getTotalDataSize() {
        Set<Map.Entry<String, IUser<T>>> es = getUserMap().entrySet();
        return es.stream().collect(Collectors.summingLong(m -> m.getValue().getTotalDataSize()));
    }

    /**
     * 添加消息
     *
     * @param key 每个用户1个key
     * @param msg 消息
     */
    public void addMsg(String key, Msg<T> msg) {
        Objects.requireNonNull(msg, "msg is null");
        Objects.requireNonNull(key, "key is null");
        msg.setUserName(key);
        addMsg(msg);
    }

    /**
     * 添加消息
     *
     * @param msg 消息
     */
    public void addMsg(Msg<T> msg) {
        Objects.requireNonNull(msg, "msg is null");
        Objects.requireNonNull(msg.getUserName(), "msg.getUserName() is null");
        IUser<T> user = getUserByKey(msg.getUserName());//不存在的用户新建一个
        user.addMsg(msg);//把消息加进用户消息队列
        userQueue.addUser(user);//把用户加进用户队列
    }


    /**
     * 根据key获取用户
     * userMap.computeIfAbsent 禁止嵌套使用就不会死循环
     *
     * @param key
     * @return
     */
    public IUser<T> getUserByKey(String key) {
        return userMap.computeIfAbsent(key, k -> createUser(k));
    }

    /**
     * 根据消息类型创建对应的用户对象
     *
     * @param key 用户标识键值
     * @return 创建的用户对象
     */
    private IUser<T> createUser(String key) {
        IUser<T> u;
        //根据消息类型枚举创建不同类型的用户对象
        switch (msgTypeEnum) {
            case TwoQueue:
                u = new TwoUser<>(key, batchSize);
                break;
            case PriorityQueue:
                u = new PriorityUser<>(key, batchSize);
                break;
            default:
                u = new OneUser<>(key, batchSize);
        }
        //设置回调函数，当恢复时，重新把用户加入队列
        Consumer<IUser<T>> biConsumer = user -> {
            logger.info("resume and add user to userQueue -> " + user.getName());
            userQueue.addUser(user);
        };
        u.setResumeCallBack(biConsumer);
        return u;
    }


    /**
     * 返回用户map
     *
     * @return
     */
    public Map<String, IUser<T>> getUserMap() {
        return userMap;
    }

    /**
     * 返回用户队列
     *
     * @return
     */
    public UserQueue<T> getUserQueue() {
        return userQueue;
    }

    public MsgTypeEnum getMsgTypeEnum() {
        return msgTypeEnum;
    }

    /**
     * 消息类型
     *
     * @param msgTypeEnum
     * @return
     */
    public TaskManage setMsgTypeEnum(MsgTypeEnum msgTypeEnum) {
        this.msgTypeEnum = msgTypeEnum;
        return this;
    }

    /**
     * 每个用户取一批的数量，默认
     *
     * @param batchSize
     * @return
     */
    public TaskManage setBatchSize(int batchSize) {
        this.batchSize = batchSize;
        return this;
    }


    /**
     * 设置重试消息过滤器
     * RetryFactory.build
     *
     * @param retryMsgFilter
     * @return
     */
    public TaskManage setRetryMsgFilter(IRetryMsgFilter retryMsgFilter) {
        this.retryMsgFilter = retryMsgFilter;
        return this;
    }

    /**
     * 设置测试模式
     *
     * @param test
     * @return
     */
    public TaskManage setTest(boolean test) {
        this.userQueue.setTest(test);
        return this;
    }

    /**
     * 运行时，改变线程池大小，注意：需要核心线程数与最大线程数一致才会生效
     *
     * @param newValueFunc
     * @return
     */
    public TaskManage bindThreadSize(IntSupplier newValueFunc) {
        isInitBindThread = true;
        this.threadSize.setValue(newValueFunc.getAsInt());
        this.threadSize.setNewValueFunc(newValueFunc);
        this.threadSize.setChangeFunc(v -> setThreadSize(v));
        return this;
    }

    /**
     * 线程数改变时调用
     * .onChangeThreadSize((newValue,oldValue)-> System.out.println("线程数由"+oldValue+"改为"+newValue))
     *
     * @param onChangeFunc(newValue,oldValue)
     * @return
     */
    public TaskManage onChangeThreadSize(BiConsumer<Integer, Integer> onChangeFunc) {
        this.threadSize.setOnChangeFunc(onChangeFunc);
        return this;
    }

    /**
     * 运行时，改变队列大小
     *
     * @param newValueFunc
     * @return
     */
    public TaskManage bindQueueCapacity(IntSupplier newValueFunc) {
        isInitBindThread = true;
        this.queueCapacity.setValue(newValueFunc.getAsInt());
        this.queueCapacity.setNewValueFunc(newValueFunc);
        this.queueCapacity.setChangeFunc(v -> setQueueCapacity(v));
        return this;
    }

    /**
     * 队列大小改变时调用
     * .onChangeQueueCapacity((newValue,oldValue)-> System.out.println("队列数由"+oldValue+"改为"+newValue))
     *
     * @param onChangeFunc(newValue,oldValue)
     * @return
     */
    public TaskManage onChangeQueueCapacity(BiConsumer<Integer, Integer> onChangeFunc) {
        this.queueCapacity.setOnChangeFunc(onChangeFunc);
        return this;
    }

    /**
     * 每10秒检查一下参数，并更新
     */
    private void refreshParam() {
        try {
            this.threadSize.check();
            this.queueCapacity.check();
        } catch (Exception e) {
            logger.info("catch Exception:" + e.getMessage(), e);
        }
    }

    public int getRefreshParamSec() {
        return refreshParamSec;
    }

    /**
     * 刷新参数时间间隔(秒)，仅在调用start方法前有效
     *
     * @param refreshParamSec
     * @return
     */
    public TaskManage<T> setRefreshParamSec(int refreshParamSec) {
        this.refreshParamSec = refreshParamSec;
        return this;
    }
}
